package com.tuozixuan.demo;

import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

public class FluxDemo {

    public static void main(String[] args) {

        // just: 通过指定序列中包含的全部元素来创建Flux对象。创建出来的Flux序列在发布这些元素之后会自动结束
        Flux.just("java", "python", "golang");

        // fromArray: 通过数组来创建Flux对象
        String[] languages = {"java", "python", "golang"};
        Flux.fromArray(languages);

        // fromIterable：通过实现了Iterable接口的对象来创建Flux对象
        List<String> languageList = Arrays.asList("java", "python", "golang");
        Flux.fromIterable(languageList);

        // empty: 创建一个不包含任何元素，只发布结束消息的序列
        Flux.empty();

        // error: 创建一个只包含错误消息的序列
        Flux.error(new RuntimeException("数据缺失"));

        // 创建一个不包含任何消息通知的序列（不会发出任何数据、错误或完成信号）
        Flux.never();

        // 创建包含从start起始的count个数量的Integer对象的序列
        Flux.range(1, 10);

        // 创建一个包含了从0开始递增的Long对象的序列。其中包含的元素按照指定的间隔来发布。除了间隔时间之外，还可以指定起始元素发布之前的延迟时间
        Flux.interval(Duration.ofMillis(5000));

        Flux.generate(sink -> {
            sink.next("java");
            sink.complete();
        }).subscribe(System.out::println);

        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) -> {
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if (list.size() == 10) {
                sink.complete();
            }
            return list;
        }).subscribe(System.out::println);

        // create: 使用的是FluxSink对象。FluxSink支持同步和异步的消息产生，并且可以在一次调用中产生多个元素。
        Flux.create(sink -> {
            for(int i = 0; i < 10; i ++)
                sink.next(i);
            sink.complete();
        }).subscribe(System.out::println);
    }
}
